package elephant.rpc.client.core;

import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import elephant.rpc.client.RPCClient;
import elephant.rpc.server.message.RPCMessage;
import elephant.rpc.server.session.RPCSession;
import elephant.utils.DumpUtil;

/**
 * 
 * @author skydu
 *
 */
public abstract class NetworkService {
	//
	private static Logger logger=LoggerFactory.getLogger(NetworkService.class);
	//
	protected RPCClient client;
	protected AtomicLong messageId;
	//
	public NetworkService(RPCClient client){
		this.client=client;
		this.messageId=new AtomicLong(1);
	}
	/**
	 * 
	 * @throws Exception
	 */
	public abstract void init() throws Exception;
	
	/**
	 * 
	 * @throws Exception
	 */
	public abstract void start() throws Exception;
	
	/**
	 * 
	 * @throws Exception
	 */
	public abstract void stop() throws Exception;
	
	/**
	 * 
	 * @param session
	 */
	public abstract void connectToServer(RPCSession session);
	
	/**
	 * 
	 * @param session
	 * @param message
	 */
	public void onMessageReceived(RPCSession session, RPCMessage message) {
		switch (message.type) {
			case RPCMessage.TYPE_RESP:
	    		onRespMessageReceive(message);
				break;
			case RPCMessage.TYPE_PUSH:
				onPushMessageReceived(session,message);
				break;
			default:
				logger.warn("bad message type {}",message);
				break;
		}
	}
	//
	private void onRespMessageReceive(RPCMessage msg){
		try {
			ResponseLock lock=client.getSyncLockMap().remove(msg.id);
			if(lock==null){
				logger.warn("lock is null bad response {}",msg);
			}else{
				lock.responseMessage=msg;
				synchronized (lock) {
					lock.notifyAll();
				}	
			}
		} catch (Exception e){
			logger.warn(e.getMessage(),e);
		}
	}
	//
	private void onPushMessageReceived(RPCSession session,RPCMessage message) {
		if(client.getPushMessageCallback()==null){
			logger.warn("pushMessageCallback is null message no process {}",DumpUtil.dump(message));
			return;
		}
		String cluster=session.serverCluster;
		String serviceId=(String)message.payloads[0];
		Object payload=message.payloads[1];
		client.getThreadPoolManager().execute(()->{
			client.getPushMessageCallback().callback(cluster,serviceId,payload);
		});
	}
	//
	protected void auth(RPCSession session){
		RPCMessage msg=new RPCMessage();
		msg.id=messageId.incrementAndGet();
		msg.type=RPCMessage.TYPE_AUTH;
		Set<String>topics=client.getTopicMap().get(session.serverCluster);
		if(topics!=null){
			session.topics=topics;
		}
		msg.payloads=new Object[2+session.topics.size()];
		msg.payloads[0]=session.clientUuid;
		msg.payloads[1]=RPCMessage.AUTH_SIGN;
		int index=2;
		for(String topic:session.topics){
			msg.payloads[index++]=topic;
		}
		session.write(msg);
		if(logger.isDebugEnabled()){
			logger.debug("auth msg {}",msg);
		}
	}
}
